This is the logic to process deletes across all log files irrespective of merge mode. This is going to be a bit costly because we're doing both merged and unmerged log records scanning to figure out the deletes.

won't this also give us deleted records from a previous log file even if not touched in this commit of interest?

I have changed the RLI logic as discussed, however, for SI the logic is same i.e. we still read from SI (to figure out delete records). But it avoids any collection on the driver. SI update is all through distributed task execution now.

Trying to see if we should optimize only to fetch keys that are part of delete log blocks here?

do we have a tracking ticket to support inserts to log files. For eg, w/ bucket index, we should be able to create secondary index. May be today, we rely on RLI. but eventually we need to support SI for any table and index type.

HUDI-8793. IMO, we should take it up with metadata DAG rewrite - #12269
Even now, I am not very happy with the double work of getting inserts/updates/deletes, one in the write handle and then again while updating metadata table. It's basically merging all log records in every commit, which neutralizes the advantages of MOR.

why can't we do count and then do repartition(count/batchSize)

Count will trigger the DAG, and repartition could trigger shuffles right? I am just doing simple transformation to batch record keys.

we need to fix L 806 to 814.

also, don't we need to group by fileSlice for all write stats?
for eg, if there are more than 1 log file added to the same file slice, does the current record generation will be right?

good point! log files can roll over and we can have more than 1 log file in the same slice. I will update this.

I thought we could simplify this.

1. parse previous log files to generate {recordKey -> isDeleted}
2. parse entire file slice (including inflight) to generate {recordKey -> isDeleted}

we are interested in deleted record keys w/ current log file. So, we could do something like

for every valid entry in (1)
   if deleted in (2) => add to deletedRecordKeyList. 

Yeah I thought about doing it that way but could not convince myself that it will always be correct for log files for all kinds of merge strategies. It works perfectly well for base files.

My line of thinking is as follows:

1. Include keys that are valid in prev and deleted in current
2. Exclude keys that are deleted in prev but revived in current
3. Include keys that are newly deleted in current

For example, let's say we have 4 keys spread across 3 log files as below. 
Here, `Prev` means 2 previous log files and `Current` means entire slice including the third inflight log file.
True/False indicates `isDeleted`, and `-` means key not present.

Key   Prev    Current
k1    false   true
k2    true    false
k3      -     true
k4    false    -

If we go by the simpler logic, then it will only consider [k1, k4] (as these two are the only valid entries in previous log files). And then finally the deletedRecordKeyList will contain only k1. However, I think the expected deleted keys should be [k1, k3] right.

not sure if we need this fix if we not getting rid of entries to be deleted from SI processed separately.

I am talking about

 // Build a list of keys that need to be removed. A 'delete' record will be emitted into the respective FileGroup of
    // the secondary index partition for each of these keys.
    HoodieData<String> keysToRemove = getRecordKeysDeletedOrUpdated(engineContext, commitMetadata, dataWriteConfig.getMetadataConfig(), dataMetaClient, instantTime);

    HoodieIndexDefinition indexDefinition = getIndexDefinition(indexPartition);
    // Fetch the secondary keys that each of the record keys ('keysToRemove') maps to
    HoodiePairData<String, String> recordKeySecondaryKeyMap =
        metadata.getSecondaryKeys(keysToRemove, indexDefinition.getIndexName(), dataWriteConfig.getMetadataConfig().getSecondaryIndexParallelism());
    HoodieData<HoodieRecord> deleteRecords = recordKeySecondaryKeyMap.map(
        (recordKeyAndSecondaryKey) -> HoodieMetadataPayload.createSecondaryIndexRecord(recordKeyAndSecondaryKey.getKey(), recordKeyAndSecondaryKey.getValue(), indexDefinition.getIndexName(), true));

In HoodieBackedTableMetadataWriter.getSecondaryIndexUpdates

lets sync up

Actually this is a remnant of one of the previous commits. you're right we may not need this. I'll check and remove if not needed.